package org.budo.dubbo.protocol.async.repository;

import java.util.Collection;
import java.util.Date;

import org.budo.dubbo.protocol.async.BudoAsyncDubboProtocol;
import org.budo.dubbo.serialize.util.DubboSerializeUtil;
import org.budo.support.lang.util.ThreadUtil;
import org.budo.time.Time;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
import com.alibaba.dubbo.common.serialize.Serialization;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invoker;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractAsyncRepository implements AsyncRepository, ApplicationContextAware, BeanNameAware, InitializingBean, DisposableBean {
    private static final String TIME_FORMAT = "yyyyMMdd.HHmmss.SSS";

    /**
     * 消息延迟超过5秒打info日志
     */
    public static final long MAX_MESSAGE_DELAY_INFO = 5 * 1000;

    /**
     * 消息延迟超过20秒打warn日志
     */
    public static final long MAX_MESSAGE_DELAY_WARN = 20 * 1000;

    /**
     * 消息延迟超过60秒打error日志
     */
    public static final long MAX_MESSAGE_DELAY_ERROR = 60 * 1000;

    public ApplicationContext applicationContext;

    /**
     * 线程启动时间
     */
    private volatile Date popThreadStartAt = null;

    /**
     * 线程结束时间
     */
    private volatile Date popThreadStopAt = null;

    /**
     * 线程是否循环跑的开关
     */
    protected boolean popThreadRun = true;

    private Thread popThread = null;

    /**
     * 空闲拉取休息时间，单位秒
     */
    private Integer popIdleSeconds = 1;

    private String beanName;

    /**
     * 如果消息延迟就打日志
     */
    protected void logMessageDelay(BudoAsyncInvocation budoAsyncInvocation) {
        if (null == budoAsyncInvocation) {
            log.error("#61 budoAsyncInvocation=" + budoAsyncInvocation + ", this=" + this);
            return;
        }

        Long createdAt = budoAsyncInvocation.getCreatedAt();
        if (null == createdAt) {
            log.debug("#66 createdAt is null, budoAsyncInvocation=" + budoAsyncInvocation + ", this=" + this);
            return;
        }

        long now = System.currentTimeMillis();
        long messageDelay = now - createdAt; // 消息延迟时间

        // 后发先至，时间不准？
        if (messageDelay < 0) {
            log.error("#79 MESSAGE_DELAY, now=" + Time.when(now).toString(TIME_FORMAT) + ", createdAt=" + Time.when(createdAt).toString(TIME_FORMAT) + ", this=" + this);
            return;
        }

        // 耗时较少，不打日志
        if (messageDelay < MAX_MESSAGE_DELAY_INFO) {
            return;
        }

        if (createdAt < now - MAX_MESSAGE_DELAY_WARN) {
            log.info("#89 MESSAGE_DELAY, now=" + Time.when(now).toString(TIME_FORMAT) + ", createdAt=" + Time.when(createdAt).toString(TIME_FORMAT) + ", this=" + this);
            return;
        }

        if (createdAt < now - MAX_MESSAGE_DELAY_ERROR) {
            log.warn("#94 MESSAGE_DELAY, now=" + Time.when(now).toString(TIME_FORMAT) + ", createdAt=" + Time.when(createdAt).toString(TIME_FORMAT) + ", this=" + this);
            return;
        }

        log.error("#82 MESSAGE_DELAY, now=" + Time.when(now).toString(TIME_FORMAT) + ", createdAt=" + Time.when(createdAt).toString(TIME_FORMAT) + ", this=" + this);
    }

    /**
     * 停止线程
     */
    @Override
    public void destroy() throws Exception {
        log.warn("#107 destroy, this=" + this);

        // 不继续循环拉数据
        this.popThreadRun = false;

        // 等待跑完当前一个
        int i = 0;
        while (null == this.popThreadStopAt) {
            i++;
            ThreadUtil.sleep(500); // 缓半秒钟

            if (i < 20 || i % 2 == 0) { // 每秒一次日志
                log.warn("#130 thread stoping, popThread=" + this.popThread);
            }
        }

        log.warn("#126 thread stoped, popThread=" + this.popThread);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.startPopMessageThread(); // 启动线程
    }

    public void startPopMessageThread() {
        if (null != this.popThreadStartAt) { // 保证每个 AsyncRepository 实例只启动一个轮训线程
            log.info("#60 popThreadStartAt=" + this.popThreadStartAt + ", return, this=" + this);
            return;
        }

        this.popThreadStartAt = new Date();

        String threadName = this.getClass().getSimpleName() + "-" + this.hashCode() + "-Thread";
        this.popThread = new Thread(this.runnable(), threadName);

        this.popThread.setDaemon(true);
        this.popThread.start();
    }

    private Runnable runnable() {
        return new Runnable() {
            public void run() {
                while (AbstractAsyncRepository.this.popThreadRun) {
                    long start = System.currentTimeMillis();

                    AbstractAsyncRepository.this._popMessage();

                    if (log.isDebugEnabled()) {
                        log.debug("#112 _popMessage, 耗时 " + (System.currentTimeMillis() - start) + ", this=" + this);
                    }
                }

                // 跳出循环时才会执行到这里
                log.warn("#164 outing thread now, this=" + this);
                AbstractAsyncRepository.this.popThreadStopAt = new Date();
            }
        };
    }

    protected void _popMessage() {
        try {
            Integer messageCount = this._doPopMessage();
            if (messageCount > 0) {

                if (log.isDebugEnabled()) {
                    log.debug("#130 _popMessage, messageCount=" + messageCount);
                }

                return;
            }

            // 如果没有消息，就休息一段时间
            Integer popIdleSeconds = this.getPopIdleSeconds();

            if (log.isDebugEnabled()) {
                log.debug("#140 sleep, messageCount=" + messageCount + ", popIdleSeconds=" + popIdleSeconds + ", this=" + this);
            }

            Thread.sleep(1000 * popIdleSeconds);
        } catch (Throwable e) {
            log.error("#145 doPopMessage error, e=" + e, e);
        }
    }

    protected Integer _doPopMessage() {
        BudoAsyncDubboProtocol dubboProtocol = BudoAsyncDubboProtocol.getInstance();

        if (null == dubboProtocol || null == dubboProtocol.getExporterMap() || dubboProtocol.getExporterMap().isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("#154 dubboPopMessage error, protocol=" + dubboProtocol + ", this=" + this);
            }

            return 0;
        }

        Collection<Exporter<?>> dubboExporters = dubboProtocol.getExporterMap().values();
        Integer messageCount = 0;

        for (Exporter<?> dubboExporter : dubboExporters) { // 每一个
            Invoker<?> invoker = dubboExporter.getInvoker();
            URL url = invoker.getUrl();
            String asyncRepositoryBeanName = url.getParameter(ASYNC_REPOSITORY_BEAN_NAME);

            if (null == asyncRepositoryBeanName || asyncRepositoryBeanName.isEmpty()) {
                log.warn("#104 asyncRepositoryBeanName is null, invoker=" + invoker);
                continue;
            }

            if (!asyncRepositoryBeanName.equals(this.getBeanName())) { // 只处理指定当前Bean的
                if (log.isDebugEnabled()) {
                    log.debug("#175 asyncRepositoryBeanName=" + asyncRepositoryBeanName + ", this=" + this);
                }

                continue;
            }

            // RpcContext.getContext().getUrl(); // 可用这个方法获取 destinationName
            messageCount += this.popMessageAndInvoke(invoker);
        }

        if (log.isDebugEnabled()) {
            log.debug("#136 dubboProtocolPopMessage, dubboExporters.size=" + dubboExporters.size() + ", " + "messageCount=" + messageCount);
        }
        return messageCount;
    }

    public BudoAsyncInvocation deserializeAsyncInvocation(byte[] messageBody, Invoker invoker) {
        URL url = null == invoker ? null : invoker.getUrl();
        Serialization serialization = this.getSerialization(url);

        return DubboSerializeUtil.deserialize(serialization, messageBody, BudoAsyncInvocation.class);
    }

    public byte[] serializeAsyncInvocation(BudoAsyncInvocation budoAsyncInvocation, Invoker<?> invoker) {
        URL url = null == invoker ? null : invoker.getUrl();
        Serialization serialization = this.getSerialization(url);

        return DubboSerializeUtil.serializeResultValue(serialization, budoAsyncInvocation, true);
    }

    /**
     * 默认 序列化方式为 java
     */
    protected Serialization getSerialization(URL url) {
        String serialization = null == url //
                ? BudoAsyncDubboProtocol.DEFAULT_SERIALIZATION //
                : url.getParameter(Constants.SERIALIZATION_KEY, BudoAsyncDubboProtocol.DEFAULT_SERIALIZATION);

        ExtensionLoader<Serialization> extensionLoader = ExtensionLoader.getExtensionLoader(Serialization.class);
        return extensionLoader.getExtension(serialization);
    }
}